#
#  Copyright 2024 The InfiniFlow Authors. All Rights Reserved.
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#

# from beartype import BeartypeConf
# from beartype.claw import beartype_all  # <-- you didn't sign up for this
# beartype_all(conf=BeartypeConf(violation_type=UserWarning))    # <-- emit warnings from all code

from api.utils.log_utils import initRootLogger
from plugin import GlobalPluginManager
initRootLogger("ragflow_server")

import logging
import os
import signal
import sys
import time
import traceback
from concurrent.futures import ThreadPoolExecutor
import threading
import uuid

from werkzeug.serving import run_simple
from api import settings
from api.apps import app
from api.db.runtime_config import RuntimeConfig
from api.db.services.document_service import DocumentService
from api import utils

from api.db.db_models import init_database_tables as init_web_db
from api.db.init_data import init_web_data
from api.versions import get_ragflow_version
from api.utils import show_configs
from rag.settings import print_rag_settings
from rag.utils.redis_conn import RedisDistributedLock

stop_event = threading.Event()

RAGFLOW_DEBUGPY_LISTEN = int(os.environ.get('RAGFLOW_DEBUGPY_LISTEN', "0"))

import debugpy; debugpy.connect(('127.0.0.1', 9509))
# 该函数用于周期性更新任务进度，使用分布式锁确保同一时间只有一个进程执行更新操作
def update_progress():
    # 生成一个唯一标识符作为锁的值，用于Redis分布式锁
    lock_value = str(uuid.uuid4())
    # 初始化一个Redis分布式锁对象，锁的名称为"update_progress"，设置超时时间为60秒
    redis_lock = RedisDistributedLock("update_progress", lock_value=lock_value, timeout=60)
    # 记录当前线程/实例持有的分布式锁的唯一标识
    logging.info(f"update_progress lock_value: {lock_value}")
    # 进入循环，持续执行进度更新任务直到stop_event被设置
    while not stop_event.is_set():
        # 尝试获取并处理任务更新逻辑
        try:
            # 尝试获取分布式锁
            if redis_lock.acquire():
                # 成功获取锁后，调用DocumentService类方法更新进度
                DocumentService.update_progress()
                # 更新完成后释放分布式锁
                redis_lock.release()
            # 每次执行完一次更新后等待6秒，控制更新频率
            stop_event.wait(6)
        # 捕获并记录执行过程中可能出现的异常
        except Exception:
            logging.exception("update_progress exception")
        # 不论是否发生异常，最终都要尝试释放锁
        finally:
            redis_lock.release()

# 信号处理函数，用于优雅地关闭程序，响应中断信号（如Ctrl+C）
def signal_handler(sig, frame):
    # 记录收到中断信号的日志信息
    logging.info("Received interrupt signal, shutting down...")
    # 设置停止事件，通知其他线程或循环终止执行
    stop_event.set()
    # 等待1秒，确保其他线程有时间响应停止事件
    time.sleep(1)
    # 退出程序，返回状态码0表示正常退出
    sys.exit(0)

# 主程序入口，用于启动RAGFlow服务并处理命令行参数、初始化配置和启动HTTP服务器
if __name__ == '__main__':
    # 打印ASCII风格的欢迎标志，增强可视化效果和调试识别
    logging.info(r"""
        ____   ___    ______ ______ __               
       / __ \ /   |  / ____// ____// /____  _      __
      / /_/ // /| | / / __ / /_   / // __ \| | /| / /
     / _, _// ___ |/ /_/ // __/  / // /_/ /| |/ |/ / 
    /_/ |_|/_/  |_|\____//_/    /_/ \____/ |__/|__/                             

    """)
    # 记录当前运行的RAGFlow版本号
    logging.info(
        f'RAGFlow version: {get_ragflow_version()}'
    )
    # 输出项目的根目录路径，便于调试或确认运行环境
    logging.info(
        f'project base: {utils.file_utils.get_project_base_directory()}'
    )
    # 显示当前配置信息，用于确认配置加载正确性
    show_configs()
    # 初始化全局设置，包括从配置文件读取配置等操作
    settings.init_settings()
    # 打印与RAG相关的设置信息，用于调试和确认
    print_rag_settings()

    # 如果启用了debugpy调试，并且监听端口已配置，则进入调试模式
    if RAGFLOW_DEBUGPY_LISTEN > 0:
        # 输出debugpy监听地址和端口信息
        logging.info(f"debugpy listen on {RAGFLOW_DEBUGPY_LISTEN}")
        # 导入debugpy模块（仅在需要时导入）
        import debugpy
        # 启动debugpy监听，允许远程调试器连接
        debugpy.listen(("0.0.0.0", RAGFLOW_DEBUGPY_LISTEN))

    # init db
    # 初始化数据库连接或结构
    init_web_db()
    # 初始化数据库初始数据（如默认用户、权限等）
    init_web_data()
    # init runtime config
    # 导入argparse模块以解析命令行参数
    import argparse

    # 创建命令行参数解析器
    parser = argparse.ArgumentParser()
    # 添加--version参数，用于显示版本信息
    parser.add_argument(
        "--version", default=False, help="RAGFlow version", action="store_true"
    )
    # 添加--debug参数，用于启用调试模式
    parser.add_argument(
        "--debug", default=False, help="debug mode", action="store_true"
    )
    # 解析命令行输入的参数
    args = parser.parse_args()
    # 如果用户输入了--version，则打印版本号并退出程序
    if args.version:
        print(get_ragflow_version())
        sys.exit(0)

    # 设置全局运行时配置的DEBUG状态
    RuntimeConfig.DEBUG = args.debug
    # 如果处于调试模式，输出日志提示
    if RuntimeConfig.DEBUG:
        logging.info("run on debug mode")

    # 初始化运行时环境变量
    RuntimeConfig.init_env()
    # 初始化运行时配置，传入主机IP和端口
    RuntimeConfig.init_config(JOB_SERVER_HOST=settings.HOST_IP, HTTP_PORT=settings.HOST_PORT)

    # 加载所有插件到全局插件管理器中
    GlobalPluginManager.load_plugins()

    # 注册SIGINT信号处理函数（通常对应Ctrl+C中断）
    signal.signal(signal.SIGINT, signal_handler)
    # 注册SIGTERM信号处理函数（通常用于优雅关闭服务）
    signal.signal(signal.SIGTERM, signal_handler)

    # 创建一个线程池执行器，最大工作线程数为1，用于异步执行任务
    thread = ThreadPoolExecutor(max_workers=1)
    # 提交update_progress任务到线程池，用于周期性更新进度
    thread.submit(update_progress)

    # 尝试启动HTTP服务器
    # start http server
    try:
        # 日志记录HTTP服务器启动信息
        logging.info("RAGFlow HTTP server start...")
        # 使用werkzeug.run_simple方法启动开发服务器
        run_simple(
            hostname=settings.HOST_IP,
            port=settings.HOST_PORT,
            application=app,
            threaded=True,
            use_reloader=RuntimeConfig.DEBUG,
            use_debugger=RuntimeConfig.DEBUG,
        )
    except Exception:
        # 出现异常时打印错误堆栈
        traceback.print_exc()
        # 设置停止事件，通知其他线程终止
        stop_event.set()
        # 等待1秒确保线程有时间响应
        time.sleep(1)
        # 强制杀死当前进程
        os.kill(os.getpid(), signal.SIGKILL)
